package com.gee.spring.cloud.data.transform.core.kafka.consumer;

import com.gee.spring.cloud.common.core.domain.frame.CstpWrapper;
import com.gee.spring.cloud.common.protobuf.ExchangeMessage;
import com.gee.spring.cloud.data.transform.core.dispatch.CenterDataDispatcher;
import com.gee.spring.cloud.data.transform.core.dispatch.BaseDataDispatcher;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author gepengjun
 * @since 2024/4/1 10:46
 */
@Component
@Slf4j
public class FrameConsumer {


    @Autowired
    private CenterDataDispatcher centerDataDispatcher;

    @Autowired
    private BaseDataDispatcher baseDataDispatcher;

    @KafkaListener(topics = {"actual_telemetry"})
    public void consumeTelemetryList(List<ExchangeMessage.Frame> frameList, Acknowledgment acknowledgment){
        try {
            frameList.forEach(this::consumerTelemetry);
        } catch (Exception e) {
            log.error("消费遥测帧数据出错", e);
        } finally {
            acknowledgment.acknowledge();
        }
    }

    private void consumerTelemetry(ExchangeMessage.Frame frame){
        ExchangeMessage.Head head = frame.getHead();
        if (baseDataDispatcher.filterOrg(head.getOrgIdentity())  //客户转发过滤的数据 不再转发给客户
                && baseDataDispatcher.filterSatellite(head.getSpacecraft())
                && baseDataDispatcher.filterDevice(head.getDevice())
                && baseDataDispatcher.filterSendBid(head.getFrameType())

        ){
            CstpWrapper cstpWrapper = CstpWrapper.parse(frame.getData().toByteArray());
            centerDataDispatcher.dispatch(cstpWrapper);
        }

    }
}
